package com.hudson.codec.domain;

import com.hudson.codec.hudson.utils.CodecUtil;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;

import java.util.Arrays;
import java.util.Map;

public class Message {

    private String topic;
    private int queueId;
    private String tags;
    private long bornTimestamp;
    private Map<String, String> properties;
    private byte[] body;

    public Message(String topic, int queueId, String tags, long bornTimestamp, Map<String, String> properties, byte[] body) {
        this.topic = topic;
        this.queueId = queueId;
        this.tags = tags;
        this.bornTimestamp = bornTimestamp;
        this.properties = properties;
        this.body = body;
    }

    public Message() {
    }

    public static Builder builder(){
        return new Builder();
    }

    // Builder 静态内部类
    public static class Builder {
        private String topic;
        private int queueId;
        private String tags;
        private long bornTimestamp;
        private Map<String, String> properties;
        private byte[] body;

        public Builder topic(String topic) {
            this.topic = topic;
            return this;
        }

        public Builder queueId(int queueId) {
            this.queueId = queueId;
            return this;
        }

        public Builder tags(String tags) {
            this.tags = tags;
            return this;
        }

        public Builder bornTimestamp(long bornTimestamp) {
            this.bornTimestamp = bornTimestamp;
            return this;
        }

        public Builder properties(Map<String, String> properties) {
            this.properties = properties;
            return this;
        }

        public Builder body(byte[] body) {
            this.body = body;
            return this;
        }

        public Message build() {
            return new Message(topic, queueId, tags, bornTimestamp, properties, body);
        }
    }

    public String getTopic() {
        return topic;
    }

    public void setTopic(String topic) {
        this.topic = topic;
    }

    public int getQueueId() {
        return queueId;
    }

    public void setQueueId(int queueId) {
        this.queueId = queueId;
    }

    public String getTags() {
        return tags;
    }

    public void setTags(String tags) {
        this.tags = tags;
    }

    public long getBornTimestamp() {
        return bornTimestamp;
    }

    public void setBornTimestamp(long bornTimestamp) {
        this.bornTimestamp = bornTimestamp;
    }

    public Map<String, String> getProperties() {
        return properties;
    }

    public void setProperties(Map<String, String> properties) {
        this.properties = properties;
    }

    public byte[] getBody() {
        return body;
    }

    public void setBody(byte[] body) {
        this.body = body;
    }

    @Override
    public String toString() {
        return "Message{" +
                "topic='" + topic + '\'' +
                ", queueId=" + queueId +
                ", tags='" + tags + '\'' +
                ", bornTimestamp=" + bornTimestamp +
                ", properties=" + properties +
                ", body=" + Arrays.toString(body) +
                '}';
    }

    /**
     * 从byte[]反序列化Message
     */
    public static Message deserialize(byte[] data) {
        if (data == null || data.length == 0) {
            return new Message();
        }

        ByteBuf byteBuf = Unpooled.wrappedBuffer(data);
        try {
            return deserialize(byteBuf);
        } finally {
            // Unpooled.wrappedBuffer创建的ByteBuf不需要释放
            // byteBuf.release(); // 注意：wrappedBuffer不需要手动释放
        }
    }

    /**
     * 序列化Message到byte[]
     */
    public static byte[] serialize(Message message) {
        if (message == null) {
            return new byte[0];
        }

        // 不提前计算长度，使用默认大小的ByteBuf
        ByteBuf byteBuf = Unpooled.buffer(); // 使用默认初始容量

        try {
            serialize(message, byteBuf);
            byte[] result = new byte[byteBuf.readableBytes()];
            byteBuf.readBytes(result);
            return result;
        } finally {
            if (byteBuf.refCnt() > 0) {
                byteBuf.release();
            }
        }
    }

    /**
     * 序列化Message到ByteBuf
     */
    public static void serialize(Message message, ByteBuf out) {
        // 序列化topic字段
        CodecUtil.writeString(out, message.topic);

        // 序列化queueId字段
        out.writeInt(message.queueId);

        // 序列化tags字段
        CodecUtil.writeString(out, message.tags);

        // 序列化bornTimestamp字段
        out.writeLong(message.bornTimestamp);

        // 序列化properties字段
        CodecUtil.writeExtData(out, message.properties);

        // 序列化body字段
        byte[] body = message.body;
        if (body == null) {
            out.writeInt(0);
        } else {
            out.writeInt(body.length);
            out.writeBytes(body);
        }
    }

    /**
     * 从ByteBuf反序列化Message
     */
    public static Message deserialize(ByteBuf in) {
        Message message = new Message();

        // 反序列化topic字段
        message.setTopic(CodecUtil.readString(in));

        // 反序列化queueId字段
        message.setQueueId(in.readInt());

        // 反序列化tags字段
        message.setTags(CodecUtil.readString(in));

        // 反序列化bornTimestamp字段
        message.setBornTimestamp(in.readLong());

        // 反序列化properties字段
        message.setProperties(CodecUtil.readExtData(in));

        // 反序列化body字段
        int bodyLength = in.readInt();
        if (bodyLength > 0) {
            byte[] body = new byte[bodyLength];
            in.readBytes(body);
            message.setBody(body);
        }

        return message;
    }
}
